package com.example.mqtt.service;

import com.example.mqtt.common.CRCnew;
import com.example.mqtt.common.JSONUtil;
import com.example.mqtt.config.RedisConfig;
import com.example.mqtt.config.Topics;
import com.example.mqtt.domain.DeviceInfoModelV4;
import com.example.mqtt.domain.RecordChangeData;
import com.example.mqtt.domain.RecordData;
import com.google.common.base.Strings;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.Message;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

@Slf4j
@Component
public class KafkaV4Component {

    @Value("${device.version.file.path}")
    private String filePath;

    @Autowired
    private RedisService redisService;

    @Autowired
    private MqttMessageGateway mqttMessageGateway;

    @Autowired
    private RecordChangeDataService recordChangeDataService;

    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @KafkaListener(topics = Topics.DEVICE_DATA_PROTO_BUF4)
    public void listenYHUBUpload(ConsumerRecord<String, byte[]> consumerRecord) {
        String topic = consumerRecord.key();
        String[] topics = topic.split("/");
        String mcuId = topics[topics.length - 1];
        DeviceInfoModelV4.YhubRequest yhubRequest4;
        byte[] byteDatas = CRCnew.chackCRC(mcuId, consumerRecord.value().clone());
        try {
            yhubRequest4 = DeviceInfoModelV4.YhubRequest.parseFrom(byteDatas);
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
            log.error("protoV4 数据解析异常");
            return;
        }
        try {
            redisService.put(RedisConfig.UPDATE_REBOOT + mcuId, false);
        } catch (Exception e) {

        }

        DeviceInfoModelV4.YhubRequest.Builder builder = DeviceInfoModelV4.YhubRequest.newBuilder();
        switch (yhubRequest4.getPayloadCase()) {
            case MBOOTREQ: // 心跳
                log.info("收到V4信息:获取实时数据 = " + mcuId);
                DeviceInfoModelV4.MessageBootReq.Builder builder1 = DeviceInfoModelV4.MessageBootReq.newBuilder();
                int serverTime = (int) (new Date().getTime() * 0.001);
                try {
                    Thread.sleep(500);
                    sendMessage(mcuId, builder.setMBootReq(builder1.setServerTime(serverTime)).build());
                } catch (Exception e) {

                }
                try {
                    DeviceInfoModelV4.MessageBootReq mBootReq = yhubRequest4.getMBootReq();
                } catch (Exception e) {
                    log.error(e.getMessage());
                }
                break;
            case MUPDATEINDEX: // 推送更新固件信息
                DeviceInfoModelV4.MessageUpdateIndex mUpdateIndex = yhubRequest4.getMUpdateIndex();
                if (mUpdateIndex.getDIsReadyUpdate()) {
                    log.info("设备准备更新 ");
                } else if (mUpdateIndex.getSUpdateFileTotalLength() != 0) {
                    redisService.remove(RedisConfig.UPDATE_INDEX + mcuId);
                    int sUpdateFileTotalNum = mUpdateIndex.getSUpdateFileTotalNum();
                    redisService.put(RedisConfig.UPDATE_INDEX + mcuId, "0/" + sUpdateFileTotalNum);
                    sendMessage(mcuId, builder.setMUpdateIndex(mUpdateIndex).build());
                }
                break;
            case MUPDATEFIRWAREPAKCETBYNUM: // 持续推送更新固件信息
                long a = System.currentTimeMillis();
                DeviceInfoModelV4.MessageUpdateFirwarePacketByNum mUpdateFirwarePakcetByNum = yhubRequest4.getMUpdateFirwarePakcetByNum();
                String fileName = redisService.get(RedisConfig.UPDATE_FILENAME + mcuId);
                byte[] fileBytes = new byte[0];
                int currentNum = mUpdateFirwarePakcetByNum.getSdPacketCurrentNum();
                try {
                    fileBytes = Files.readAllBytes(Paths.get(filePath, fileName, fileName + currentNum));
                    sendMessage(mcuId, builder.setMUpdateFirwarePakcetByNum(mUpdateFirwarePakcetByNum
                            .toBuilder()
                            .setSPacket(ByteString.copyFrom(fileBytes))
                            .setSIsPacketIn(true)
                            .build())
                            .build()
                    );

                    String indexCache = redisService.get(RedisConfig.UPDATE_INDEX + mcuId);
                    String[] progress = indexCache.split("\\/");
                    HashMap hashMap = new HashMap();
                    hashMap.put("index", currentNum);
                    hashMap.put("updateTime", System.currentTimeMillis());
                    hashMap.put("updateVersion", mUpdateFirwarePakcetByNum.getSPacket());
                    Double progressDB = ((currentNum + 1) / (Integer.parseInt(progress[1]) * 1d)) * 100;
                    hashMap.put("progress", progressDB.intValue());
                    hashMap.put("rebooting", false);
                    if (progressDB.intValue() >= 99) {
                        hashMap.put("rebooting", true);
                        redisService.put(RedisService.RedisType.UPDATESTATE + ":" + mcuId, JSONUtil.objectToJsonString(hashMap), 120l);
                    } else {
                        redisService.put(RedisService.RedisType.UPDATESTATE + ":" + mcuId, JSONUtil.objectToJsonString(hashMap));
                    }

                    break;
                } catch (IOException e) {
                    DeviceInfoModelV4.YhubRequest build = builder.setMUpdateFirwarePakcetByNum(mUpdateFirwarePakcetByNum.toBuilder()
                            .setSPacket(ByteString.copyFrom(fileBytes))
                            .setSIsPacketIn(false)
                            .build()
                    ).build();
                    sendMessage(mcuId, build);
                }

                break;
            case MUPLOADRECORDREQ:// 获取到上传非实时
                log.info("收到V4信息:获取到上传非实时 = " + mcuId);
                DeviceInfoModelV4.MessageUploadRecordReq mUploadRecordReq = yhubRequest4.getMUploadRecordReq();
                List<DeviceInfoModelV4.Record> dRecordList = mUploadRecordReq.getDRecordList();
                if (dRecordList.size() != 0) {
                    log.info("收到V4信息:准备保存record = " + mcuId);
                    for (int i = 0; i < dRecordList.size(); i++) {
                        DeviceInfoModelV4.Record record = dRecordList.get(i);
                        testChargeDisCharge(mcuId, record);
                        try {



                            /*
                            recordDataService.save(recordData);
                               */
                        } catch (Exception e) {

                            e.printStackTrace();
                        }
                    }
                }
                break;
            case MGETCONFIG: // 获取当前配置信息
                DeviceInfoModelV4.MessageGetConfig mGetConfig = yhubRequest4.getMGetConfig();
                DeviceInfoModelV4.MessageUpdateConfig getConfig = mGetConfig.getGetConfig();
                if (getConfig == null || "".equals(getConfig.toString())) {
                    sendMessage(mcuId, builder.setMGetConfig(mGetConfig).build());
                } else {
                    // 电池曲线
                    DeviceInfoModelV4.MessageBattery sdBattery = getConfig.getSdBattery();

                    // 支持液位
                    boolean isSupportWaterlevel = getConfig.getIsSupportWaterlevel();

                    try {
                        threadPoolTaskExecutor.execute(new Runnable() {
                            @Override
                            public void run() {
                                log.debug("异步保存yhub的配置");
//                                deviceConfigClient.saveDeviceConfigVo(deviceConfigVO);
                            }
                        });
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }
                break;
            case MREALTIMEDATA: // 获取实时数据
                DeviceInfoModelV4.MessageGetRealTimeData mRealTimeData = yhubRequest4.getMRealTimeData();
                log.info("收到V4信息:获取实时数据 = " + mcuId);

                try {
                    new Thread(new Runnable() {
                        @Override
                        public void run() {

                            log.info("收到V4信息:准备存入数据 = " + mcuId);
                            try {
//                                recordTableClient.save(recordTable);
//                                iRecordTableService.save(recordTable);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }

                        }
                    }).start();

                } catch (Exception e) {
                    log.error(e.getMessage());
                }

                try {
                    DeviceInfoModelV4.Record record = mRealTimeData.getDRecord();

                } catch (Exception e) {
                    log.error(e.getMessage());
                }

                break;
            case MUPDATECONFIG:
                DeviceInfoModelV4.MessageUpdateConfig updateConfig = yhubRequest4.getMUpdateConfig();
                if (updateConfig.getConfigFrom() == 1) {
                    sendMessage(mcuId, builder.setMUpdateConfig(updateConfig).build());
                } else {
                    redisService.put(RedisConfig.UPDATE_CONFIG_STATUS + mcuId, 1);
                }
                break;
            case MDEVICECONTROL:
                DeviceInfoModelV4.MessageDeviceControl mDeviceControl = yhubRequest4.getMDeviceControl();
                if (!mDeviceControl.getDIsOk()) {
                    sendMessage(mcuId, builder.setMDeviceControl(mDeviceControl).build());
                } else {
                    log.info("MDEVICECONTROL -- > 控制状态:{}", mDeviceControl.getDIsOk());
                }
                break;
            case PAYLOAD_NOT_SET:
                log.error("PAYLOAD_NOT_SET");
                break;
        }
    }

    private void testChargeDisCharge(String mcuId, DeviceInfoModelV4.Record currentRecord) {
        String chargeKey = "charge_" + mcuId;
        String disChargeKey = "disCharge_" + mcuId;
        String isBeginChargeKey = "beginCharg_" + mcuId;

        String booleanString = redisService.get(isBeginChargeKey);

        Boolean isBeginCharge = true;
        if (booleanString != null) {
            isBeginCharge = Boolean.parseBoolean(booleanString);
        }
        if (isBeginCharge) {
            RecordData tempBegin = redisService.getRecordData(chargeKey);// 从缓存取出来
            if (currentRecord.getCurrentCharging() > 1) {
                if (tempBegin != null) {
                    if ((currentRecord.getYhubStime() - tempBegin.getYhubStime()) > 5 * 60) {
                        // TODO 保存一条数据 tempBean
                        RecordChangeData recordChangeData = new RecordChangeData();
                        BeanUtils.copyProperties(tempBegin, recordChangeData);
                        recordChangeDataService.saveRecordChangeData(recordChangeData);

                        isBeginCharge = false;
                        // 缓存 isBeginCharge
                        redisService.put(isBeginChargeKey, isBeginCharge.toString());
                        // remove tempBegin
                        redisService.remove(chargeKey);
                    }
                } else {
                    tempBegin = new RecordData();
                    BeanUtils.copyProperties(currentRecord, tempBegin);
                    // TODO 缓存tempBegin
                    redisService.put(chargeKey, tempBegin);
                }
            }

        } else {
            RecordData tempEnd = redisService.getRecordData(disChargeKey);// 从缓存取出来
            if (currentRecord.getCurrentCharging() < 1) {
                if (tempEnd != null) {
                    if ((currentRecord.getYhubStime() - tempEnd.getYhubStime()) > 5 * 60) {
                        // TODO 保存一条数据 tempEnd
                        RecordChangeData recordChangeData = new RecordChangeData();
                        BeanUtils.copyProperties(tempEnd, recordChangeData);
                        recordChangeDataService.saveRecordChangeData(recordChangeData);

                        isBeginCharge = true;
                        // 缓存 isBeginCharge
                        redisService.put(isBeginChargeKey, isBeginCharge.toString());
                        // remove tempEnd
                        redisService.remove(disChargeKey);
                    }
                } else {
                    tempEnd = new RecordData();
                    BeanUtils.copyProperties(currentRecord, tempEnd);
                    // TODO 缓存tempEnd
                    redisService.put(disChargeKey, tempEnd);
                }
            }
        }
    }

    public <T> void sendMessage(String mcuId, T payload) {
        Message<byte[]> publishMsg = MessageBuilder.withPayload(CRCnew.toCRCByteV4((byte[]) payload))
                //发送的主题
                .setHeader(MqttHeaders.TOPIC, "gpd/s/04/" + mcuId).build();
        try {
            mqttMessageGateway.sendMessage(publishMsg);
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("发送消息完成:{}", publishMsg.getPayload());
    }
}
